package com.longtengyuyun;

import com.sun.xml.internal.ws.client.sei.ResponseBuilder;
import org.apache.kafka.clients.admin.*;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @author ZhaoYunLong
 * @date 2022/3/11
 * @description
 */
public class KafkaTopicDML {


    public static void main(String[] args) {
        try {
//        创建topic
            Properties props = new Properties();
            props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "10.10.60.119:9092,10.10.60.120:9092,10.10.60.121:9092");
            KafkaAdminClient adminClient = (KafkaAdminClient) KafkaAdminClient.create(props);
            createTopics(adminClient);
            listTopics(adminClient);
            System.out.println("-------------------------");
            topicInfos(adminClient);
            System.out.println("-------------------------");
            deleteTopic(adminClient);
            listTopics(adminClient);


        }catch (Exception e) {
            e.printStackTrace();
        }

    }

    private static void deleteTopic(KafkaAdminClient adminClient) throws ExecutionException, InterruptedException {
//        Short replicationFactor = 3;
//        NewTopic newTopic = new NewTopic("topic_zyl", 3, replicationFactor);
        DeleteTopicsResult topic_zyl = adminClient.deleteTopics(Arrays.asList("topic_zyl"));
        topic_zyl.all().get();
    }

    private static void createTopics(KafkaAdminClient adminClient) throws ExecutionException, InterruptedException {
        Short replicationFactor = 3;
        NewTopic newTopic = new NewTopic("topic_zyl", 3, replicationFactor);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        // 同步创建
        topics.all().get();
    }

    private static void  listTopics(KafkaAdminClient adminClient) throws ExecutionException, InterruptedException {
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        Set<String> strings = listTopicsResult.names().get();
        System.out.println(strings.size());
        strings.forEach(topicName->{
            System.out.println(topicName);
        });
    }

    private static void topicInfos(KafkaAdminClient adminClient) throws ExecutionException, InterruptedException {
        String topicName = "topic_zyl";
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(topicName));
        Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
        stringTopicDescriptionMap.forEach((name, topicSDescription)->{
            System.out.println(name + ": "+ topicSDescription.toString());
        });
    }

}
